package com.onlyxiahui.framework.net.session.server;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * Date 2018-12-30 13:19:45<br>
 * Description
 *
 * @author XiaHui<br>
 * @since 1.0.0
 */
public class DataQueueHandlerThread {

	protected final Logger logger = LoggerFactory.getLogger(this.getClass());
	private final ConcurrentLinkedQueue<DataItem> queue = new ConcurrentLinkedQueue<>();
	private ExecutorService pool;
	private final DataAction dataAction;

	private long sleepTime = 1;
	/**
	 * 非核心线程闲置时超时1s
	 */
	private static final long KEEP_ALIVE = 1000;

	public DataQueueHandlerThread(int count, DataAction dataAction) {
		this.dataAction = dataAction;
		initialize(count);
	}

	private void initialize(int count) {
		initialize(count, count);
	}

	private void initialize(int coreSize, int maxSize) {
		pool = new ThreadPoolExecutor(
				// 核心线程数量
				coreSize,
				// 最大线程数量，线程池所容纳最大线程数(workQueue队列满了之后才开启)
				maxSize,
				// 当非核心线程闲置空闲时，保持活跃的时间
				KEEP_ALIVE,
				// keepAliveTime的单位 ，毫秒级
				TimeUnit.MILLISECONDS,
				// 线程任务等待队列，存储还未执行的任务
				new LinkedBlockingQueue<>(),
				// 创建线程的工厂
				new ThreadFactoryBuilder().setNameFormat("task-pool-%d").build(),
				// 由于达到线程边界或队列容量而阻塞时的处理程序
				new ThreadPoolExecutor.AbortPolicy());
		for (int i = 0; i < coreSize; i++) {
			pool.execute(new DataHandleThread());
		}
	}

	public long getSleepTime() {
		return sleepTime;
	}

	public void setSleepTime(long sleepTime) {
		this.sleepTime = sleepTime;
	}

	private void handle(DataItem dataQueue) {
		if (null != dataAction) {
			dataAction.handle(dataQueue);
		}
	}

	public void push(DataItem dataItem) {
		queue.add(dataItem);
	}

	public int size() {
		return queue.size();
	}

	class DataHandleThread extends Thread {
		@Override
		public void run() {
			DataItem dataQueue;
			while (true) {
				while (!queue.isEmpty()) {
					dataQueue = queue.poll();
					handle(dataQueue);
				}
				threadSleep(sleepTime);
			}
		}

		private void threadSleep(long time) {
			try {
				sleep(time);
			} catch (InterruptedException e) {
				logger.error("", e);
			}
		}
	}
}
